package cartagows.wsframework.wsatomictransaction;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map.Entry;

import org.apache.axiom.om.OMElement;
import org.apache.axis2.addressing.EndpointReference;

import cartago.ArtifactId;
import cartago.CartagoException;
import cartago.Op;
import cartago.OpFeedbackParam;
import cartago.OpRequestTimeoutException;
import cartagows.CartagowsParam;
import cartagows.WSMsgInfo;
import cartagows.util.WSAgent;
import cartagows.util.XMLib;
import cartagows.wsframework.core.FaultBuilder;
import cartagows.wsframework.core.ServiceConfiguration;
import cartagows.wsframework.wsaddressing.WSAddressingMessageElem;
import cartagows.wsframework.wscoordination.WSCoordinationErrorParam;
import cartagows.wsframework.wscoordination.WSCoordinationMessageElem;
import cartagows.wsframework.wscoordination.WSCoordinationParam;

public class ATCoordinatorAgent extends WSAgent{

	private String activityId;
	private ArrayList<String> durable2PCParticipants;
	private ArrayList<String> volatile2PCParticipants;
	private boolean volatileParticipantFinished;
	private boolean isTransactionToCommit;
	private HashMap<String,String> participantList;
	private int participantsEnded;
	private String ACTIVE="ACTIVE";
	private String PREPARING="PREPARING";
	private String PREPARED="PREPARED";
	private String COMMITTING="COMMITTING";
	private String COMMITTED="COMMITTED";
	private String ABORTING="ABORTING";
	private String ABORTED="ABORTED";
	private String currentState;
	private ArtifactId wsProxyParticipant; 
	private ArtifactId wsProxyInitiator; 
	private int finalParticipantNumber;
	private OMElement coordContext;
	private ArtifactId activitiesRegistry;
	private ArtifactId messagesWarehouse;
	
	public ATCoordinatorAgent(String agentName, String activityId) throws CartagoException{

		super(agentName);

		this.activityId = activityId;
		durable2PCParticipants = new ArrayList<String>();
		volatileParticipantFinished = false;
		isTransactionToCommit = true;
		finalParticipantNumber = 0;
		currentState = ACTIVE;
		volatile2PCParticipants = new ArrayList<String>();
		durable2PCParticipants = new ArrayList<String>();
		
		activitiesRegistry = lookupArtifact(WSCoordinationParam.ACTIVITIES_REGISTRY_DEF_NAME);
		messagesWarehouse = lookupArtifact(WSCoordinationParam.ACTIVITIES_MSG_WAREHOUSE_DEF_NAME);
	}
	
	public void run(){
		
		String opName ="";
		Entry<String,String> entry= null;
		String senderEPR = ""; 
		String initiatorEPR ="";
		EndpointReference targetEPR = null;

		//Creation of the WSInterface to use with the participants 
		targetEPR = null;
		try {
			wsProxyParticipant = createWSInterface(WSAtomicTransactionParam.WSINTERFACE_DEF_NAME+activityId,
					WSAtomicTransactionParam.WSDL_URI, WSAtomicTransactionParam.PARTICIPANT_PORT, targetEPR);
		
			//The local EPR is setted to WSAtomicTransactionParam.DEFAULT_COORD_EPR (the ATCoordinator
			//that dispatch the messages to the AgentCoordinatorAT)
			doAction(wsProxyParticipant,new Op("setLocalEPR", WSAtomicTransactionParam.DEFAULT_COORD_EPR));

		}
		catch (Exception ex) {
			log("[ATCOORDINATOR_AGENT:]Error preparing the artifacts");
			ex.printStackTrace(); 
		}

		log("[ATCOORDINATOR_AGENT:]Starting");
		
		while(true){
			try {
				
				//Waiting request
				log("[ATCOORDINATOR_AGENT:]Waiting request");
				OpFeedbackParam<WSMsgInfo> msgFeedback = new OpFeedbackParam<WSMsgInfo>();
				doAction(messagesWarehouse, new Op("getNextMessage", activityId, msgFeedback));
				
				log("[ATCOORDINATOR_AGENT:]Request found");
				
				WSMsgInfo inputMsg = msgFeedback.get();
				opName = inputMsg.getOperationName();
				
				OMElement header = XMLib.getInstance().buildElementFromString(inputMsg.getHeader());
				senderEPR = XMLib.getInstance().getElementValue(WSAddressingMessageElem.REPLYTO, WSAddressingMessageElem.ADDRESS, header);
				if(coordContext==null){
					coordContext = XMLib.getInstance().getElement(WSCoordinationMessageElem.COORDINATION_CONTEXT, header);
					//Configuration of the artifact. From now, all the messages sent with the wsProxy just
					//configured will contain the coordination context coordContext
					doAction(wsProxyParticipant,new Op("configure", new ServiceConfiguration(coordContext)));
				}
				
				if(participantList==null){
					OpFeedbackParam<HashMap<String,String>> participantsFeedback = new OpFeedbackParam<HashMap<String,String>>();
					doAction(activitiesRegistry, new Op("getActivityParticipantList",activityId, participantsFeedback), CartagowsParam.LONG_TIMEOUT);
					participantList = participantsFeedback.get();
				}
	
				//CommitOperation
				if (opName.equals(WSAtomicTransactionWSDLOperation.COMMIT)) {
					
					if(participantList.containsKey(senderEPR)){
					
						log("[ATCOORDINATOR_AGENT:]Commit operation found");
						//The sender of the commit message is the initiator
						doAction(activitiesRegistry, new Op("setInitiator",activityId,senderEPR));
						initiatorEPR = senderEPR;
						participantList.remove(initiatorEPR);
						//Creation and configuration of the WSInterface that will be used for the communication
						//with the initiator
						initProxyInitiator(new EndpointReference(initiatorEPR));
						
						//Check if the transaction is in the right state
						if(currentState.equals(ACTIVE)){
							
							log("[ATCOORDINATOR_AGENT:]Transaction in PREPARING state");
							
							currentState = PREPARING;
							Iterator<Entry<String,String>>it = participantList.entrySet().iterator();
							int volatileCount = 0;
							
							//Send PREPARE message to the participants registered with the VOLATILE_2PC protocol
							log("[ATCOORDINATOR_AGENT:]Sending prepare message to the VOLATILE_2PC participants");
							while(it.hasNext()){
								entry = it.next();
								if(entry.getValue().equals(WSAtomicTransactionParam.VOLATILE_2PC)){
									volatile2PCParticipants.add(entry.getKey());
									//Set the targetEPR
									doAction(wsProxyParticipant,new Op("setTargetEPR", entry.getKey()));
									//Send message to the participant
									doOneWay(wsProxyParticipant, WSAtomicTransactionWSDLOperation.PREPARE,
											createPrepareMessage().toString(), CartagowsParam.LONG_TIMEOUT);
									volatileCount++;
								}
								else
									durable2PCParticipants.add(entry.getKey());
							}
							//Send the PREPARE message to the participants registered with the DURABLE_2PC protocol
							//(No VOLATILE_2PC participants found)
							if(volatileCount==0){
								log("[ATCOORDINATOR_AGENT:]No VOLATILE_2PC participants registered. " +
										"Sending prepare messagge to the DURABLE_2PC participants");
								volatileParticipantFinished = true;
								Iterator<String> it2 = durable2PCParticipants.iterator();
								while(it2.hasNext()){
									//Set the targetEPR
									doAction(wsProxyParticipant, new Op("setTargetEPR", it2.next()));
									//Send message to the participant
									doOneWay(wsProxyParticipant, WSAtomicTransactionWSDLOperation.PREPARE,
											createPrepareMessage().toString(), 
											CartagowsParam.LONG_TIMEOUT);
								}
							}
						}
						//Message invalid for the current state of the coordinator
						else{
							log("[ATCOORDINATOR_AGENT:]Error: Message invalid for the current state of the coordinator");
							sendInvalidStateFault(senderEPR);
						}
					}
				}
				//Rollback operation
				else if(opName.equals(WSAtomicTransactionWSDLOperation.ROLLBACK)) {
					if(participantList.containsKey(senderEPR)){
						
						log("[ATCOORDINATOR_AGENT:]Rollback operation found");
						
						if(currentState.equals(ACTIVE)){
							isTransactionToCommit = false;
	
							//The sender of the rollback message is the initiator
							doAction(activitiesRegistry, new Op("setInitiator",activityId,senderEPR));
							
							initiatorEPR = senderEPR;
							participantList.remove(initiatorEPR);
							//Creation and configuration of the WSInterface that will be used for the communication
							//with the initiator
							initProxyInitiator(new EndpointReference(initiatorEPR));

							log("[ATCOORDINATOR_AGENT:]Transaction in ABORTING state");
							
							currentState = ABORTING;
							//Send of the ROLLBACK message to all the participants
							log("[ATCOORDINATOR_AGENT:]Sending the rollback message to all the participants");
							
							Iterator<Entry<String,String>>it = participantList.entrySet().iterator();
							while(it.hasNext()){
								entry = it.next();
								if(!entry.equals(senderEPR)){
									//Set the targetEPR
									doAction(wsProxyParticipant,new Op("setTargetEPR", entry.getKey()));
									//Send message to the participant
									doOneWay(wsProxyParticipant, WSAtomicTransactionWSDLOperation.ROLLBACK,
											createRollbackMessage().toString(), CartagowsParam.LONG_TIMEOUT);
								}
							}
							finalParticipantNumber = participantList.size();
						}
						//Message invalid for the current state of the coordinator
						else{
							log("[ATCOORDINATOR_AGENT:]Error: Message invalid for the current state of the coordinator");
							sendInvalidStateFault(senderEPR);
						}
					}
				}
				//Prepared operation
				else if (opName.equals(WSAtomicTransactionWSDLOperation.PREPARED)){
					if(participantList.containsKey(senderEPR)){
						
						log("[ATCOORDINATOR_AGENT:]Prepared operation found");
						
						if(currentState.equals(PREPARING)){

							finalParticipantNumber = participantList.size();
							//Check if there are VOLATILE_2PC participants
							//that still have to respond to the coordinator
							
							if(!volatileParticipantFinished){

								Iterator<String> it = volatile2PCParticipants.iterator();
								String currentEPR ="";
								while(it.hasNext()){
									currentEPR = it.next();
									if(currentEPR.equals(senderEPR))
										it.remove();
								}
								if(volatile2PCParticipants.size()==0){
									log("[ATCOORDINATOR_AGENT:]All the VOLATILE_2PC participant has terminated the Prepare phase");
									volatileParticipantFinished = true;
									//Send of the PREPARE message to all the DURABLE_2PC participants								
									if(durable2PCParticipants.size()!=0){
										log("[ATCOORDINATOR_AGENT:]Sending the prepare message to the DURABLE_2PC participants");
										it = durable2PCParticipants.iterator();
										while(it.hasNext()){
											//Set the targetEPR
											doAction(wsProxyParticipant,new Op("setTargetEPR", entry.getKey()));
											//Send message to the participant
											doOneWay(wsProxyParticipant, WSAtomicTransactionWSDLOperation.PREPARE, 
													createPrepareMessage().toString(),CartagowsParam.LONG_TIMEOUT);
										}
									}
									//Send of the COMMIT message to the participants registered with the DURABLE_2PC protocol
									//(No VOLATILE_2PC participants found)
									else{
										
										log("[ATCOORDINATOR_AGENT:]Transaction in PREPARED state");
										currentState = PREPARED;
 
										Iterator<Entry<String,String>>it2 = participantList.entrySet().iterator();
										while(it2.hasNext()){
											entry = it2.next();
											//Set the targetEPR
											doAction(wsProxyParticipant,new Op("setTargetEPR", entry.getKey()));
											//Send message to the participant
											doOneWay(wsProxyParticipant, WSAtomicTransactionWSDLOperation.COMMIT,
													createCommitMessage().toString(), CartagowsParam.LONG_TIMEOUT);
										}
										
										log("[ATCOORDINATOR_AGENT:]Transaction in COMMITTING state");
										
										currentState = COMMITTING;
									}
								}
							}
							//Received a PREPARED message from a DURABLE_2PC participant
							else{
								Iterator<String> it = durable2PCParticipants.iterator();
								while(it.hasNext()){
									if(it.next().equals(senderEPR))
										it.remove();
								}
								//Check if all the DURABLE_2PC participant has sent their reply to the coordinator 
								if(durable2PCParticipants.size()==0){
									
									log("[ATCOORDINATOR_AGENT:]Transaction in PREPARED state");
									currentState = PREPARED;

									//Send of the COMMIT message to all the participants 
									Iterator<Entry<String,String>>it2 = participantList.entrySet().iterator();
									while(it2.hasNext()){
										entry = it2.next();
										//Set the targetEPR
										doAction(wsProxyParticipant,new Op("setTargetEPR", entry.getKey()));
										//Send message to the participant
										doOneWay(wsProxyParticipant, WSAtomicTransactionWSDLOperation.COMMIT,
												createCommitMessage().toString(), CartagowsParam.LONG_TIMEOUT);
									}

									log("[ATCOORDINATOR_AGENT:]Transaction in COMMITTING state");

									currentState = COMMITTING;
								}
							}
						}
						else{
							log("[ATCOORDINATOR_AGENT:]Error: Message invalid for the current state of the coordinator");
							sendInvalidStateFault(senderEPR);
						}
					}
				}

				//Readonly operation: the participant has voted for the commit of the transaction. However the
				//participant dosen't want to be involved in the second phase of the 2PC protocol
				else if (opName.equals(WSAtomicTransactionWSDLOperation.READ_ONLY)){
					if(participantList.containsKey(senderEPR)){
						if(currentState.equals(PREPARING)){
							//Removal of the participant from the participants list
							participantList.remove(senderEPR);
							finalParticipantNumber = participantList.size();
						}
						else{
							log("[ATCOORDINATOR_AGENT:]Error: Message invalid for the current state of the coordinator");
							sendInvalidStateFault(senderEPR);
						}
					}
				}
				//Aborted operation
				else if (opName.equals(WSAtomicTransactionWSDLOperation.ABORTED)){
					if(participantList.containsKey(senderEPR) || initiatorEPR.equals(senderEPR)){
						if(currentState.equals(PREPARING) || currentState.equals(ABORTING)){
							log("[ATCOORDINATOR_AGENT:]Aborted operation found");
							//If isTransactionToCommit = true it means that the message just received is
							//the first AbortedOperation the coordinator has encountered
							if(isTransactionToCommit){

								log("[ATCOORDINATOR_AGENT:]Transaction in ABORTING state");
								isTransactionToCommit = false;
								currentState = ABORTING;
	
								Iterator<Entry<String,String>>it = participantList.entrySet().iterator();
								//Send the ROLLBACK message to all the participants
								log("[ATCOORDINATOR_AGENT:]Sending ROLLBACK message to all the participants");
								while(it.hasNext()){
									entry = it.next();
									//Set the targetEPR
									doAction(wsProxyParticipant,new Op("setTargetEPR", entry.getKey()));
									//Send message to the participant
									doOneWay(wsProxyParticipant, WSAtomicTransactionWSDLOperation.ROLLBACK, 
											createRollbackMessage().toString(), CartagowsParam.LONG_TIMEOUT);
								}
								if(participantList.containsKey(senderEPR)){
									participantList.remove(senderEPR);
									finalParticipantNumber = participantList.size();
								}
							}
							//Acquire of the rollback message from all the participants. At the end
							//the initiator will be informed of the transaction rollback
							else{
								if(participantList.containsKey(senderEPR)){
									
									participantsEnded++;
									participantList.remove(senderEPR);
									//Check if all the participants have replied to the coordinator with an
									//aborted message
									if(participantsEnded == finalParticipantNumber){

										log("[ATCOORDINATOR_AGENT:]All the participants have rollbacked the transaction");
										
										OpFeedbackParam<String> initiatorFeedback = new OpFeedbackParam<String>();
										doAction(activitiesRegistry,new Op("getInitiatorEPR",activityId, initiatorFeedback));
										initiatorEPR = initiatorFeedback.get();
										
										log("[ATCOORDINATOR_AGENT:]Transaction in ABORTED state");
										//Send the transaction outcome to the initiator
										doOneWay(wsProxyInitiator, WSAtomicTransactionWSDLOperation.ABORTED,
												createAbortedMessage().toString(), CartagowsParam.LONG_TIMEOUT);
										currentState = ABORTED;
										//Removal of the transaction info
										removeTransactionInfo(activityId);
										log("[ATCOORDINATOR_AGENT:]Transaction successfully ABORTED");
										//The agent has finished the processing regarding this transaction, 
										//so it ends its execution
										break;
									}
								}
							}
						}
						else{
							log("[ATCOORDINATOR_AGENT:]Error: Message invalid for the current state of the coordinator");
							sendInvalidStateFault(senderEPR);
						}
					}
				}
				//Committed operation
				else if (opName.equals(WSAtomicTransactionWSDLOperation.COMMITED)){
					if(participantList.containsKey(senderEPR)){
						if(currentState.equals(COMMITTING)){
							participantsEnded++;
							participantList.remove(senderEPR);

							//Check if all the participants have replied to the coordinator with a
							//COMMITTED message
							if(participantsEnded==finalParticipantNumber){
								
								log("[ATCOORDINATOR_AGENT:]All the participants have COMMITTED the transaction");
								OpFeedbackParam<String> initiatorFeedback = new OpFeedbackParam<String>();
								doAction(activitiesRegistry,new Op("getInitiatorEPR", activityId, initiatorFeedback));
								initiatorEPR = initiatorFeedback.get();
								
								//Send the transaction outcome to the initiator
								doOneWay(wsProxyInitiator, WSAtomicTransactionWSDLOperation.COMMITED,
										createCommittedMessage().toString(),CartagowsParam.LONG_TIMEOUT);
								currentState = COMMITTED;
								//Removal of the transaction info
								removeTransactionInfo(activityId);
								log("[ATCOORDINATOR_AGENT:]Transacion successfully committed");
								//The agent has finished the processing regarding this transaction, 
								//so it ends its execution
								break;
							}
						}
						else{
							log("[ATCOORDINATOR_AGENT:]Error: Message invalid for the current state of the coordinator");
							sendInvalidStateFault(senderEPR);
						}
					}
				}
			}
			catch(OpRequestTimeoutException ex){
				log("[ATCOORDINATOR_AGENT:]No msg");
			}
			catch(Exception ex){
				log("[ATCOORDINATOR_AGENT:]"+ ex.getMessage() +" "+ ex.getCause());
				ex.printStackTrace();
			}
		}
	}
	
	private OMElement createPrepareMessage(){
		OMElement msg = XMLib.getInstance().buildElement(WSCoordinationMessageElem.PREPARE);
	    return msg;
	}
	
	private OMElement createRollbackMessage(){
		OMElement msg = XMLib.getInstance().buildElement(WSCoordinationMessageElem.ROLLBACK);
		return msg;
	}

	private OMElement createCommitMessage(){
		OMElement msg = XMLib.getInstance().buildElement(WSCoordinationMessageElem.COMMIT);
		return msg;
	}

	private OMElement createCommittedMessage(){
		OMElement msg = XMLib.getInstance().buildElement(WSCoordinationMessageElem.COMMITTED);
	    return msg;
	}
	
	private OMElement createAbortedMessage(){
		OMElement msg = XMLib.getInstance().buildElement(WSCoordinationMessageElem.ABORTED);
	    return msg;
	}

	private void initProxyInitiator(EndpointReference initiatorEPR) throws Exception{
		wsProxyInitiator = createWSInterface(WSAtomicTransactionParam.WSINTERFACE_DEF_NAME+activityId+"2",
				WSAtomicTransactionParam.WSDL_URI, WSAtomicTransactionParam.INITIATOR_PORT_TYPE, initiatorEPR);
		doAction(wsProxyInitiator, new Op("setLocalEPR", WSAtomicTransactionParam.DEFAULT_COORD_EPR));
	}
	
	private void removeTransactionInfo(String activityId) throws Exception{
		doAction(activitiesRegistry, new Op("removeActivity",activityId));		
		doAction(messagesWarehouse, new Op("removeActivity",activityId));
	}
	
	private void sendInvalidStateFault(String endPoint) throws Exception{
		
		OMElement msg = FaultBuilder.getInstance().buildSOAPFault(
				WSCoordinationErrorParam.FAULT_INCONSISTENT_STATE_CODE, 
				WSCoordinationErrorParam.FAULT_INCONSISTENT_STATE_SUBCODE, 
				WSCoordinationErrorParam.FAULT_INCONSISTENT_STATE_REASON, "");

		//Send the fault message to the EPR that caused the fault
		doAction(wsProxyParticipant,new Op("setTargetEPR", endPoint.toString()));
		doOneWay(wsProxyParticipant,CartagowsParam.FAULT_OP,
				msg.toString(), CartagowsParam.LONG_TIMEOUT);
		
		Iterator<Entry<String,String>>it = participantList.entrySet().iterator();
		Entry<String,String> entry;
		//Send the rollback message to all the participants
		while(it.hasNext()){
			entry = it.next();
			//Set the targetEPR
			doAction(wsProxyParticipant,new Op("setTargetEPR", entry.getKey()));
			//Send message to the participant
			doOneWay(wsProxyParticipant, WSAtomicTransactionWSDLOperation.ROLLBACK , 
					createRollbackMessage().toString(), CartagowsParam.LONG_TIMEOUT);
		}
	}
}